package flink_p2_sql

import org.apache.flink.api.common.typeinfo.TypeInformation
import org.apache.flink.streaming.api.scala.{StreamExecutionEnvironment, createTypeInformation}
import org.apache.flink.table.api.{DataTypes, Table, Types}
import org.apache.flink.table.api.scala.StreamTableEnvironment
import org.apache.flink.table.functions.TableFunction
import org.apache.flink.types.Row


object tableApi04_tableApi_udf {

  /**
   * @param args
   */
  def main(args: Array[String]): Unit = {



    val streamEnv: StreamExecutionEnvironment = StreamExecutionEnvironment.getExecutionEnvironment

    val tableEnv: StreamTableEnvironment = StreamTableEnvironment.create(streamEnv)



    import org.apache.flink.streaming.api.scala._
    import org.apache.flink.table.api.scala._

    val socketStream: DataStream[String] = streamEnv.socketTextStream("127.0.0.1", 8889)



    //将DataStream[String] 转换为 table并指定列名叫 line
    val table: Table = tableEnv.fromDataStream(socketStream, 'line)


    val myFun = new myFlatMapFunction


    // test1 : 输出myFun 的结果
    val res1: Table = table
      .flatMap(myFun('line)).as('word,'word_count)      // myFun 输出的是两列，所以这里需要as两个字段名
      .select('word, 'word_count)

    //test2 : 输出myFun 的结果 并按照word聚合（有状态）

    val table1: Table = table.flatMap(myFun('line)).as('word, 'word_count)
      .groupBy('word)
      .select('word, 'word_count.sum)


    // 第一种方式得到DataStream
    val resStream1: DataStream[(Boolean, Row)] = res1.toRetractStream[Row].filter(_._1)

    //第二种方式得到DataStream
    val resStream2: DataStream[(Boolean, Row)] = tableEnv.toRetractStream[Row](table1)
      .filter(_._1)


    resStream2.print().setParallelism(2)


    tableEnv.execute("test")
  }
}

/**
 * 通过udf实现一行数据按空格分隔并输出 (key，1）类型
 */

class myFlatMapFunction extends TableFunction[Row] {
  override def getResultType: TypeInformation[Row] = {
    //import org.apache.flink.table.api.Types
    Types.ROW(Types.STRING, Types.INT)
  }


  def eval(line: String) = {
    val arr: Array[String] = line.split(" ")
    val row = new Row(2)
    for (elem <- arr) {
      row.setField(0, elem)
      row.setField(1, 1)
      collect(row)                      //发射到下游
    }
  }
}